package cc.eddic.practice.concurrency.impl;

import cc.eddic.practice.concurrency.ExecutorServiceUtil;
import cc.eddic.practice.concurrency.LogItem;
import cc.eddic.practice.concurrency.LogParser;
import lombok.extern.slf4j.Slf4j;
import lombok.val;

import java.io.File;
import java.io.IOException;
import java.nio.file.Files;
import java.nio.file.Paths;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.stream.Collectors;
import java.util.stream.Stream;

import static cc.eddic.practice.concurrency.FileUtil.gzipReader;

@Slf4j
public class LogParserWithExecuterImpl implements LogParser {
    LogItemParserLhImpl logItemParserLh = new LogItemParserLhImpl();

    @Override
    public Map<String, List<LogItem>> parse(String folderName) {
        Map<String, List<LogItem>> map = new ConcurrentHashMap<>();
        List<File> filesList = new ArrayList<>();
        fileResult(new File(folderName), filesList);
        ExecutorService executorService= Executors.newFixedThreadPool(16);
        CountDownLatch latch=new CountDownLatch((int) filesList.stream().filter(file -> file.getName().endsWith(".gz")).count());
        filesList.stream().filter(file -> file.getName().endsWith(".gz"))
                .forEach(f -> executorService.submit(()-> mapAdd(f,map,latch)));
        try {
            latch.await();
        }catch (InterruptedException e){
            log.error(e.getMessage());
        }
        ExecutorServiceUtil.shutdown(executorService);
        return map;
    }
    private void mapAdd(File f, Map<String, List<LogItem>> map, CountDownLatch latch)  {
        LogItemGZipParser logItemGZipParser = new LogItemGZipParser();
//        if (log.isDebugEnabled())log.debug("filename  {}",f.getName());
        try {
            List<LogItem> logItemList=gzipReader(f.getPath()).map(logItemGZipParser::parse)
                    .filter(Optional::isPresent).map(Optional::get).collect(Collectors.toList());
            latch.countDown();
            map.put(f.getName(), logItemList);
        }catch (IOException E){
            log.error(E.getMessage());
        }
//        List<LogItem> logItemList=readList(f.getPath()).stream().map(filestring -> logItemParserLh.parse(filestring))
//                .filter(Optional::isPresent).map(Optional::get).collect(Collectors.toList());

    }
    //遍历文件夹
    private void fileResult(File file, List<File> filesList) {
        val files = file.listFiles();
        if (files == null) {
            return;
        }
        for (val f : files) {
            if (f.isDirectory()) {
                fileResult(f, filesList);
            } else filesList.add(f);
        }
    }

    public Stream<String> read(String textFileName) {
        try {
            val p = Paths.get(textFileName);
            return Files.lines(p);
        } catch (IOException e) {
            if (log.isDebugEnabled()) log.debug("{}", textFileName);
            return null;
        }
    }

    public List<String> readList(String textFileName) {
        return read(textFileName).collect(Collectors.toList());

    }
}
